package com.asap.demo.sink;

import com.asap.demo.ck.handle.ProcessForCk;
import com.asap.demo.utils.PluginContext;
import com.asap.rule.StandardEvent;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;


/**
 * @author wangbh
 * @Description: ClickhouseSink
 * @date 2021/11/30 14:53
 */
public class ClickhouseSink extends RichSinkFunction<StandardEvent> {
    private static final Logger logger = LoggerFactory.getLogger(ClickhouseSink.class);
    ProcessForCk processForCk = new ProcessForCk();
    private final List<StandardEvent> localValues = new ArrayList<>();

    private int maxFlushBufferSize = 0;
    private long timeoutMillis = 0L;

    private volatile long lastAddTimeMillis = 0L;
    public boolean firstFlag = true;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        PluginContext context = new PluginContext();
        context.setConfigPath("/home/asap/wbh/");
        processForCk.initialize(context);
        maxFlushBufferSize = 5000;
        timeoutMillis = 60000;
    }

    @Override
    public void invoke(StandardEvent value, Context context) throws Exception {
        if (firstFlag) {
            // 当前批次第一次进入，记录开始时间
            firstFlag = false;
            lastAddTimeMillis = System.currentTimeMillis();
        }
        put(value);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    private void addToQueue() throws Exception {
        List<StandardEvent> deepCopy = buildDeepCopy(localValues);
        processForCk.processMsg(deepCopy);
        localValues.clear();
        firstFlag = true;
    }

    public void put(StandardEvent value) throws Exception {
        tryAddToQueue();
        localValues.add(value);
    }

    synchronized void tryAddToQueue() throws Exception {
        if (flushCondition()) {
            addToQueue();
        }
    }

    private boolean flushCondition() {
        return localValues.size() > 0 && (checkSize() || checkTime());
    }

    private boolean checkSize() {
        return localValues.size() >= maxFlushBufferSize;
    }

    private boolean checkTime() {
        long current = System.currentTimeMillis();
        return current - lastAddTimeMillis > timeoutMillis;
    }

    private static List<StandardEvent> buildDeepCopy(List<StandardEvent> original) {
        return Collections.unmodifiableList(new ArrayList<>(original));
    }
}
